package com.example.hhhh;

import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.widget.Button;

import com.alibaba.android.arouter.facade.annotation.Route;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

import static java.lang.Thread.sleep;

/**
 * @ProjectName: Hhhh
 * @Package: com.example.hhhh
 * @ClassName: FristAct
 * @Description: java类作用描述
 * @Author: songhua
 * @CreateDate: 2019/12/20 0020 16:02
 */
@Route(path = "/app/SixAct")
public class SixAct extends SwipeBackActivity implements View.OnClickListener {
    private static final String TAG = "SixAct";
    private Button bt1, bt2, bt3, bt4, bt5, bt6;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_six);
        init();
    }

    private void init() {
        bt1 = (Button) findViewById(R.id.bt1);
        bt2 = (Button) findViewById(R.id.bt2);
        bt3 = (Button) findViewById(R.id.bt3);
        bt4 = (Button) findViewById(R.id.bt4);
        bt5 = (Button) findViewById(R.id.bt5);
        bt6 = (Button) findViewById(R.id.bt6);
        bt1.setOnClickListener(this);
        bt2.setOnClickListener(this);
        bt3.setOnClickListener(this);
        bt4.setOnClickListener(this);
        bt5.setOnClickListener(this);
        bt6.setOnClickListener(this);
    }

    @Override
    public void onClick(View v) {
        switch (v.getId()) {
            case R.id.bt1:
                //第一步：创建连载小说（被观察者）
                Observable novel = Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        emitter.onNext("连载1");
                        emitter.onNext("连载2");
                        emitter.onNext("连载3");
                        emitter.onComplete();
                    }
                });
                //第二步：创建读者（观察者）
                Observer<String> reader = new Observer<String>() {
                    private Disposable mDisposable;
                    @Override
                    public void onSubscribe(Disposable d) {
                        mDisposable = d;
                        Log.e(TAG, "onSubscribe");
                    }

                    @Override
                    public void onNext(String value) {
                        if ("2".equals(value)) {
                            mDisposable.dispose();
                            return;
                        }
                        Log.e(TAG, "onNext:" + value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError=" + e.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        Log.e(TAG, "onComplete()");
                    }
                };
                //第三步：读者和连载小说建立订阅关系
                novel.subscribe(reader);//一行代码搞定
                break;
            case R.id.bt2:
                Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        emitter.onNext("连载1");
                        emitter.onNext("连载2");
                        emitter.onNext("连载3");
                        emitter.onComplete();
                    }
                })
                        .observeOn(AndroidSchedulers.mainThread())//回调在主线程
                        .subscribeOn(Schedulers.io())//执行在io线程
                        .subscribe(new Observer<String>() {
                            @Override
                            public void onSubscribe(Disposable d) {
                                Log.e(TAG,"onSubscribe");
                            }

                            @Override
                            public void onNext(String value) {
                                Log.e(TAG,"onNext:"+value);
                            }

                            @Override
                            public void onError(Throwable e) {
                                Log.e(TAG,"onError="+e.getMessage());
                            }

                            @Override
                            public void onComplete() {
                                Log.e(TAG,"onComplete()");
                            }
                        });

                break;
            case R.id.bt3:
                Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        emitter.onNext(123);
                        sleep(3000);
                        emitter.onNext(456);
                    }
                }).observeOn(AndroidSchedulers.mainThread())
                        .subscribeOn(Schedulers.io())
                        .subscribe(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) throws Exception {
                                Log.e(TAG,integer+"");
                            }
                        }, new Consumer<Throwable>() {
                            @Override
                            public void accept(Throwable throwable) throws Exception {

                            }
                        }, new Action() {
                            @Override
                            public void run() throws Exception {

                            }
                        });

                break;
            case R.id.bt4:
                Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe( ObservableEmitter<Integer> e) throws Exception {
                        Log.e(TAG, "Observable emit 1" + "\n");
                        e.onNext(1);
                        Log.e(TAG, "Observable emit 2" + "\n");
                        e.onNext(2);
                        Log.e(TAG, "Observable emit 3" + "\n");
                        e.onNext(3);
                        e.onComplete();
                        Log.e(TAG, "Observable emit 4" + "\n" );
                        e.onNext(4);
                    }
                }).subscribe(new Observer<Integer>() {
                    private int i;
                    private Disposable mDisposable;

                    @Override
                    public void onSubscribe( Disposable d) {
                        Log.e(TAG, "onSubscribe : " + d.isDisposed() + "\n" );
                        mDisposable = d;
                    }

                    @Override
                    public void onNext( Integer integer) {
                        Log.e(TAG, "onNext : value : " + integer + "\n" );
                        i++;
                        if (i == 2) {
                            // 在RxJava 2.x 中，新增的Disposable可以做到切断的操作，让Observer观察者不再接收上游事件
                            mDisposable.dispose();
                            Log.e(TAG, "onNext : isDisposable : " + mDisposable.isDisposed() + "\n");
                        }
                    }

                    @Override
                    public void onError( Throwable e) {
                        Log.e(TAG, "onError : value : " + e.getMessage() + "\n" );
                    }

                    @Override
                    public void onComplete() {
                        Log.e(TAG, "onComplete" + "\n" );
                    }
                });


                break;
            case R.id.bt5:

                Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe( ObservableEmitter<Integer> e) throws Exception {
                        e.onNext(1);
                        e.onNext(2);
                        e.onNext(3);
                    }
                }).map(new Function<Integer, String>() {
                    @Override
                    public String apply( Integer integer) throws Exception {
                        return "This is result " + integer;
                    }
                }).subscribe(new Consumer<String>() {
                    @Override
                    public void accept( String s) throws Exception {
                        Log.e(TAG, "accept : " + s +"\n" );
                    }
                });
                //map 基本作用就是将一个 Observable 通过某种函数关系，转换为另一种 Observable，
                // 上面例子中就是把我们的 Integer 数据变成了 String 类型
                break;
            case R.id.bt6:


                break;

        }

    }
}
